package com.atguigu.app.func;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.bean.TableProcess;
import com.atguigu.util.JdbcUtil;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.Connection;
import java.sql.DriverManager;
import java.util.*;

public class DwdTableProcessFunction extends BroadcastProcessFunction<JSONObject, String, JSONObject> {
    private MapStateDescriptor<String, TableProcess> mapStateDescriptor;
    private HashMap<String, TableProcess> tableProcessHashMap;

    public DwdTableProcessFunction(MapStateDescriptor<String, TableProcess> mapStateDescriptor) {
        this.mapStateDescriptor = mapStateDescriptor;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        tableProcessHashMap = new HashMap<>();
        Connection connection = DriverManager.getConnection( "jdbc:mysql://47.102.112.46:3306/edu_config?" +
                "user=root&password=qw123456&useUnicode=true&" +
                "characterEncoding=utf8&serverTimeZone=Asia/Shanghai&useSSL=false" );
        List<TableProcess> tableProcesses = JdbcUtil.queryList( connection,
                "select * from table_process where sink_type='dwd'",
                TableProcess.class,
                true );
        for (TableProcess tableProcess : tableProcesses) {
            String key = tableProcess.getSourceTable() + tableProcess.getSourceType();
            tableProcessHashMap.put( key,tableProcess );
        }
    }

    @Override  // TODO: 2023/3/27 广播流  获取数据 过滤数据 转为javabean对象  写入状态
    public void processBroadcastElement(String value, Context ctx, Collector<JSONObject> out) throws Exception {
        BroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState( mapStateDescriptor );
        JSONObject jsonObject = JSONObject.parseObject( value );
        if ("d".equals( jsonObject.getString( "op" ) )) {
            JSONObject before = jsonObject.getJSONObject( "before" );
            String source_table = before.getString( "source_table" );
            String source_type = before.getString( "source_type" );
            String key = source_table + "-" + source_type;
            broadcastState.remove( key );
            tableProcessHashMap.remove( key );
        } else {
            // todo 解析数据
            TableProcess tableProcess = JSON.parseObject( jsonObject.getString( "after" ), TableProcess.class );

            // todo 将数据写入状态
            String key = tableProcess.getSourceTable() + "-" + tableProcess.getSourceType();
            broadcastState.put( key, tableProcess );
        }
    }

    @Override  // TODO: 2023/3/27 主流  根据广播流过滤数据   进行 行 列 过滤   写出状态备用
    public void processElement(JSONObject value, ReadOnlyContext ctx, Collector<JSONObject> out) throws Exception {
        ReadOnlyBroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState( mapStateDescriptor );
        String key = value.getString( "table" ) + "-" + value.getString( "type" );
        TableProcess tableProcess = broadcastState.get( key );
        TableProcess tableProcessMap = tableProcessHashMap.get( key );

        if (tableProcess != null || tableProcessMap!=null)  {
            if (tableProcess==null){
                tableProcess=tableProcessMap;
            }
            filterColumns( value.getJSONObject( "data" ), tableProcess.getSinkColumns() );

            // TODO: 2023/3/27 补充sinkTable 字段写出
            value.put( "sink_table", tableProcess.getSinkTable() );
            out.collect( value );
        } else {
            System.out.println( "没有该key的信息" + key );
        }
    }

    private void filterColumns(JSONObject data, String sinkColumns) {
        Set<Map.Entry<String, Object>> entries = data.entrySet();
        Iterator<Map.Entry<String, Object>> iterator = entries.iterator();

        String[] fields = sinkColumns.split( "," );
        List<String> fieldList = Arrays.asList( fields );

        while (iterator.hasNext()) {
            Map.Entry<String, Object> entry = iterator.next();
            if (!fieldList.contains( entry.getKey() )) {
                iterator.remove();
            }
        }
    }


}
